package com.headcaselabs.queue;

import java.util.Collection;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.List;
import java.util.PriorityQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.xerox.amazonws.sqs.SQSException;


public class WeightedRoundRobin implements QueueSelectionAlgorithm {
	
	private PriorityQueue<QueueParameters> queueOfQueues;
	private Hashtable<String,QueueParameters> queuename_parameters = new Hashtable<String,QueueParameters>();
	//private PriorityQueue<QueueParameters> queueOfQueues;
	private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

  private final static Log log_ =
    LogFactory.getLog(WeightedRoundRobin.class.getName());
	
	public WeightedRoundRobin(Collection<QueueParameters> queues, Object lock) {
		queueOfQueues = new PriorityQueue<QueueParameters>(queues);
		for (QueueParameters qp : queues) {
			queuename_parameters.put(qp.getQueueName(),qp);
		}
	}
	
	
	public QueueParameters nextQueue() {
			List<QueueParameters> stack = new LinkedList<QueueParameters>();
			try {
				lock.writeLock().lock();
				QueueParameters toReturn = queueOfQueues.poll();
				//stack.add(toReturn);
				int i = 0;
				try {
					if (toReturn!=null)
						log_.debug("Approximate number of messages in queue:"+toReturn.getQueueName()+" #"+toReturn.getQueue().getApproximateNumberOfMessages());
					while (toReturn !=null && toReturn.getQueue().getApproximateNumberOfMessages() <= 0) {
						stack.add(toReturn);
						toReturn = queueOfQueues.poll();
						i++;
						if (toReturn!=null)
							log_.debug("Approximate number of messages in queue:"+toReturn.getQueueName()+" #"+toReturn.getQueue().getApproximateNumberOfMessages());
					}
				} catch (SQSException sqse) {
					for (QueueParameters p : stack) {
						queueOfQueues.add(p);
					}
					log_.error("Error: SQS threw an exception. Restoring Queue order...");
					log_.error(sqse);
				}
				if (toReturn !=null) {
					toReturn.addExTime();
					queueOfQueues.add(toReturn);
					log_.debug("Returned Queue:"+toReturn.getQueueName());
				} else {
					log_.debug("No messages found in any queues. Iterated over:"+i+" queues.");
				}
				for (QueueParameters p : stack) {
					if (p!=null) {
						queueOfQueues.add(p);
						log_.debug("Adding queue:"+p+" back to stack");
					}
				}
				log_.debug("Number of Queues in Queue:"+queueOfQueues.size());
				return toReturn;
		} finally {
			lock.writeLock().unlock();
		}
		
	}


	public synchronized void  updateQueueParameters(String queue, long actualTime) {
		//try {
			//lock.writeLock().lock();
			log_.info("Removing queue from queues");
			QueueParameters qp = queuename_parameters.get(queue);
			queueOfQueues.remove(qp);
			log_.info("Adjusting Cpu Time");
			qp.adjustCpuTime(actualTime);
			log_.info("Adding queue tp queues");
			queueOfQueues.add(qp);
			log_.info("Added queue to queues - done");
			
		//} finally {
			//lock.writeLock().unlock();
		//}
	}
	
	public void printQueues() {
		try {
			lock.readLock().lock();
			for (QueueParameters p : queueOfQueues) {
				System.out.println(p.toString());
			}
		} finally {
			lock.readLock().unlock();
		}
	}


	public int getQueueCount() {
		try {
			lock.readLock().lock();
			return queueOfQueues.size();
		} finally {
			lock.readLock().unlock();
		}
	}
	
	
	

}
